package rpc.service;

import com.alibaba.fastjson.JSON;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import rpc.annotation.Demotion;
import rpc.common.NodeInformation;
import rpc.common.TaskRpc;
import rpc.init.RpcConfig;
import rpc.init.ServiceInit;
import rpc.units.SocketUnits;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
//@Component
public class Service implements ApplicationRunner,ApplicationContextAware {

    /**
     * 不足点: 想使用SpringAop,但是被先前socket调用的低级编写方式:[java的反射调用]给击败了,不得不向反射低头。
     */
    /**
     * 后序进阶版:
     * UserServiceImpl user = applicationContext.getBean("user", UserServiceImpl.class);
     * 获取对应的bean,即可融入springAop,实现aop式调用。降低耦合。
     */

    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("服务端启动");
        log.info("ip:"+rpcConfig.getIp()+";端口:"+rpcConfig.getPort());

        Map<String, Object> demotionBeanMap = applicationContext.getBeansWithAnnotation(Demotion.class);
        Map<String,Object> classifyBeanMap=new HashMap<>();
        for (Map.Entry<String, Object> entry : demotionBeanMap.entrySet()) {
            Demotion demotion = entry.getValue().getClass().getAnnotation(Demotion.class);
            String classify = demotion.classify();
            classifyBeanMap.put(classify,entry.getValue());
            if(demotionItemMap.containsKey(classify)){
                throw new RuntimeException("一个服务不可设置多个熔断");
            }
            demotionItemMap.put(classify,new DemotionItem(demotion.classify(),demotion.timeRange(),demotion.threshold(),demotion.coolingTime(),demotion.timeOut()));
        }
        this.demotionBeanMap =classifyBeanMap;

        this.registerServer();
        this.initServer();
    }

    private ApplicationContext applicationContext;

    @Autowired
    private RpcConfig rpcConfig;

    /**
     * 问: 是否复用第一个请求过去的socket;
     * 答: 只有serverSocket才可以高并发,所以并不可以;
     */
    private Map<String,Object> beanServerMap = ServiceInit.serverBeanMap;
    private ExecutorService threadPool = Executors.newFixedThreadPool(5);

    /**
     * 服务端不仅要 接收 还需要发送
     */

    //前往注册中心进行注册
    @SneakyThrows
    private void registerServer(){

        Socket socket=new Socket();
        SocketAddress socketAddress=new InetSocketAddress(rpcConfig.getRegistryCenterIp(), rpcConfig.getRegistryCenterPort());

        try {
            socket.connect(socketAddress);//连接失败直接停止任务
        }catch (Exception e){
            log.error("注册中心连接失败");
            e.printStackTrace();
            System.exit(0);
            return;
        }

        NodeInformation nodeInformation=new NodeInformation(rpcConfig.getServerName(),rpcConfig.getIp(),rpcConfig.getPort(),1);
        //发送自己的节点信息
        SocketUnits.sendSuccessMsg(socket,JSON.toJSONString(nodeInformation));
        //问: 不关闭,用于注册中心检测心跳行不行? 答:那要多开一个线程监听数据,还不如让server解决
        this.close(socket);//注册完毕就关闭
    }

    //启动服务
    @SneakyThrows
    private void initServer(){
        ServerSocket serverSocket=new ServerSocket();
        SocketAddress socketAddress=new InetSocketAddress(rpcConfig.getIp(), rpcConfig.getPort());
        serverSocket.bind(socketAddress);

        while (true){

            Socket socket = serverSocket.accept();


            threadPool.execute(()->{
                try {
                    String msg = SocketUnits.receiveMsg(socket);//接收任务

                    if(msg.equals("jump")){
                        //注册中心的心跳,返回自己的节点信息
                        SocketUnits.sendSuccessMsg(socket, JSON.toJSONString(new NodeInformation(rpcConfig.getServerName(), rpcConfig.getIp(),rpcConfig.getPort(),1)));
                        return;
                    }

                    TaskRpc taskRpc = JSON.parseObject(msg, TaskRpc.class);

                    String classify = taskRpc.getClassify();

                    Object o = this.demotionAround(taskRpc);
                    if(o!=null){
                        //熔断触发
                        SocketUnits.sendSuccessMsg(socket,JSON.toJSONString(o));
                        return;
                    }

                    Object bean = beanServerMap.get(classify);

                    if(bean==null){
                        SocketUnits.sendErrorMsg(socket,"没有对应classify");
                        log.error("没有对应classify");
                        return;
                    }

                    String methodName = taskRpc.getMethodName();
                    Method method = bean.getClass().getMethod(methodName,taskRpc.getParamsClass());

                    if(method==null){
                        SocketUnits.sendErrorMsg(socket,"没有对应方法");
                        log.error("没有对应方法");
                        return;
                    }

                    //起始时间
                    long startTime=System.currentTimeMillis();
                    //方法执行
                    Object result = method.invoke(bean, taskRpc.getParams());

                    //模拟随机
//                    Thread.sleep(new Random().nextInt(7000));

                    //返回时间
                    DemotionItem demotionItem = demotionItemMap.get(classify);
                    //拥有降级服务
                    if(demotionItem!=null) {
                        if ((System.currentTimeMillis() - startTime) / 1000 >= demotionItem.getTimeOut()) {
                            //已经超时了
                            System.out.println("超时一次");
                            demotionItem.addTimeCount();
                        }
                    }
                    String sendMsg = JSON.toJSONString(result);
                    SocketUnits.sendSuccessMsg(socket,sendMsg);//发送结果
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    //防止连接丢失 或者 客户端不主动关闭
                    try {
                        this.close(socket);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });

        }
    }

    private void close(Socket socket) throws IOException {
        socket.close();
    }

    //分界线

    private Map<String,Object> demotionBeanMap =new HashMap<>();
    private Map<String, DemotionItem> demotionItemMap=new HashMap<>();


    public Object demotionAround(TaskRpc taskRpc) {

        if(demotionItemMap.containsKey(taskRpc.getClassify())){
            //拥有熔断
            return haveDemotion(demotionItemMap.get(taskRpc.getClassify()),taskRpc);
        }else{
            //无熔断
            return null;
        }

    }

    private Object haveDemotion(DemotionItem demotionItem,TaskRpc taskRpc){

        if(demotionItem.isCooling()){
            //已经进入冷却了,触发熔断
            Object bean = demotionBeanMap.get(demotionItem.getClassify());

            Method demotionMethod=null;
            try {
                demotionMethod = bean.getClass().getMethod(taskRpc.getMethodName(),taskRpc.getParamsClass());
            } catch (NoSuchMethodException e) {
                log.error("降级类中没有对应方法");
                e.printStackTrace();
                return null;
            }
            Object result=null;
            try {
                result = demotionMethod.invoke(bean,taskRpc.getParams());
            } catch (IllegalAccessException | InvocationTargetException e) {
                log.error("调用方法失败");
                e.printStackTrace();
                return null;
            }

            return result;
        }


        //没有触发熔断
        return null;

    }

    @Data
    class DemotionItem{

        public DemotionItem(String classify,int timeRange, double threshold, int coolingTime, int timeOut) {
            this.classify=classify;
            this.timeRange = timeRange;
            this.threshold = threshold;
            this.coolingTime = coolingTime;
            this.timeOut = timeOut;
            this.allCount=new AtomicInteger();
            this.outCount=new AtomicInteger();
        }

        public boolean isCooling(){

            allCount.incrementAndGet();

            if(this.cooling &&System.currentTimeMillis()>(startTime+coolingTime*1000)){
                //冷却时间已到
                this.cooling =false;
                this.startTime=System.currentTimeMillis();
                allCount.set(0);
                outCount.set(0);
            }

            if(!this.cooling){
                if(this.startTime+(timeRange*1000)<System.currentTimeMillis()){
                    //超出了时间范围,全部置零
                    this.startTime=System.currentTimeMillis();
                    this.allCount.set(0);
                    this.outCount.set(0);
                }
            }

            return this.cooling;
        }

        public void addTimeCount(){

            int count = this.outCount.incrementAndGet();
            if(((double)count/(double)allCount.get())>=this.threshold){
                this.cooling =true;
                this.startTime=System.currentTimeMillis();//设置冷却起始时间
            }

        }

        private String classify;

        //服务是否可用
        private boolean cooling;

        //计时的开始时间/冷却的开始时间
        private long startTime;

        //总次数
        private AtomicInteger allCount;

        //当前的超时次数
        private AtomicInteger outCount;

        //时间范围
        private int timeRange;

        //次数百分比阈值
        private double threshold;

        //冷却时间
        private int coolingTime;

        //最长容忍时间
        private int timeOut;

    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext=applicationContext;
    }

}
